package com.company.poker.channel;

import akka.actor.*;
import org.jboss.netty.channel.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Description of class.
 * Created 4/10/13 4:46 PM
 *
 * @author andrey.rodin@playtech.com
 */

public class ClientHandler extends SimpleChannelHandler {
    private static final Logger logger = LoggerFactory.getLogger(ClientHandler.class);
    private ActorRef proxy;
    private ActorRef table;
    private ActorSystem system;
    private ConcurrentHashMap<SocketAddress, Channel> channelCash = new ConcurrentHashMap<SocketAddress, Channel>();

    public ClientHandler(ActorRef proxy) {
        this.proxy = proxy;

    }

    public ClientHandler() {

    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        super.messageReceived(ctx, e);
        if (!channelCash.contains(e.getChannel())) {
            channelCash.put(e.getRemoteAddress(),e.getChannel());
        }
        String request = (String) e.getMessage();
        logger.info(this.toString() + " "+request);
        proxy.tell(e.getRemoteAddress()+"->"+request + '\n', proxy);
//        proxy.tell("Send to Actor ->");
    }

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        super.channelConnected(ctx, e);
        logger.debug("host " + e.getChannel().getRemoteAddress() + " is connected");
        e.getChannel().write("[" + e.getChannel().getRemoteAddress() + "] " + "Hello" + '\n');
    }

    @Override
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        super.channelDisconnected(ctx, e);
        logger.debug("host " + e.getChannel().getRemoteAddress() + " is disconnected");
    }

    @Override
    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        super.writeRequested(ctx, e);
    }

    public ActorRef getProxy() {
        return proxy;
    }

    public void setProxy(ActorRef proxy) {
        this.proxy = proxy;
    }

    public ActorSystem getSystem() {
        return system;
    }

    public ConcurrentHashMap<SocketAddress, Channel> getChannelCash() {
        return channelCash;
    }

    public void setSystem(ActorSystem system) {
        this.system = system;
        this.proxy = system.actorOf(new Props(new UntypedActorFactory() {
            @Override
            public Actor create() {
                return new Proxy();
            }
        }),"proxy");
        this.table = system.actorOf(new Props(TableActor.class));
    }



    public class Proxy extends UntypedActor {

        public Proxy getProxy() {
            return this;
        }

        @Override
        public void onReceive(Object message) {
            String msg = (String) message;
            System.out.println("Proxy - " + msg);
            if (msg.startsWith("Table")) {
                // TODO: For test get first channel
//                Channel ch = (Channel) getChannelCash().values().toArray()[0];
                for (Channel channel : channelCash.values()) {
                    channel.write(message);
                }
//                ch.write(message);
            } else {
                table.tell(msg, getSelf());
            }

        }
    }
}
